import asyncio
import ssl
import threading
import time
from collections import deque

import websockets
from websockets.exceptions import ConnectionClosed

from ..utils.logger import get_logger
from ..eztrade_client import PyMessage, Authority


logger = get_logger("eztrade.ws", level="warning")


class WebSocketClient:
    """Async WebSocket client mirroring the WebTransport client API."""

    __slots__ = (
        "connection",
        "server",
        "port",
        "path",
        "buffer",
        "stop_flag",
        "connected",
        "bg_thread",
        "bg_loop",
        "_receive_task",
        "_connect_future",
        "_last_message_ts",
        "_secure",
    )

    def __init__(self, buffer_size: int = 65_536, *, secure: bool = True) -> None:
        self.connection: websockets.ClientConnection | None = None
        self.server: str | None = None
        self.port: int | None = None
        self.path: str | None = None
        self.buffer: deque[bytes] = deque(maxlen=buffer_size)
        self.stop_flag = asyncio.Event()
        self.connected = False
        self.bg_thread: threading.Thread | None = None
        self.bg_loop: asyncio.AbstractEventLoop | None = None
        self._receive_task: asyncio.Task | None = None
        self._connect_future: asyncio.Future | None = None
        self._last_message_ts: float = time.time()
        self._secure = secure

    async def connect(self, server: str, port: int, path: str, user_id: int | None = None) -> None:
        """Establish WebSocket connection and keep receiving until stopped."""

        self.server = server
        self.port = port
        self.path = path
        uri = self._build_uri(server, port, path)
        ssl_context = self._build_ssl_context() if self._secure else None
        logger.info("Connecting to WebSocket server %s ...", uri)

        try:
            async with websockets.connect(
                uri,
                ssl=ssl_context,
                ping_interval=20,
                ping_timeout=10,
                close_timeout=5,
                max_size=None,
            ) as websocket:
                self.connection = websocket
                self.connected = True
                self.stop_flag.clear()
                logger.debug("WebSocket connection established")

                await self._send_authority(user_id)
                self._receive_task = asyncio.create_task(self._receive_loop())
                await self._receive_task
        except asyncio.CancelledError:
            raise
        except ConnectionClosed as exc:
            logger.warning("WebSocket connection closed: code=%s reason=%s", exc.code, exc.reason)
        except Exception as exc:  # noqa: BLE001
            logger.error("WebSocket client error: %s", exc)
        finally:
            await self._cleanup_connection()

    async def _receive_loop(self) -> None:
        """Consume WebSocket frames and store binary payloads in buffer."""

        assert self.connection is not None
        websocket = self.connection
        self._last_message_ts = time.time()

        while not self.stop_flag.is_set():
            try:
                message = await asyncio.wait_for(websocket.recv(), timeout=30)
            except asyncio.TimeoutError:
                logger.debug("No message for 30s, sending ping")
                try:
                    await websocket.ping()
                except ConnectionClosed:
                    logger.info("Ping failed, connection already closed")
                    break
                except Exception as exc:  # noqa: BLE001
                    logger.error("Ping failed: %s", exc)
                    break
                continue
            except ConnectionClosed:
                logger.info("WebSocket connection closed by server")
                break
            except asyncio.CancelledError:
                raise
            except Exception as exc:  # noqa: BLE001
                logger.error("Receive loop error: %s", exc)
                break

            if message is None:
                logger.info("Received empty message, closing connection")
                break

            self._last_message_ts = time.time()

            if isinstance(message, str):
                logger.debug("Received text frame, converting to bytes")
                data = message.encode()
            else:
                data = message

            if data:
                self.buffer.append(data)

        self.stop_flag.set()

    async def _send_authority(self, user_id: int | None) -> None:
        if user_id is None:
            return
        try:
            await self._send_bytes(PyMessage(Authority(user_id)).to_binary_bytes())
        except Exception as exc:  # noqa: BLE001
            logger.error("Send authority message failed: %s", exc)
            raise

    async def _send_bytes(self, payload: bytes) -> None:
        if not self.connected or self.connection is None:
            raise RuntimeError("WebSocket session is not established")
        if self.stop_flag.is_set():
            raise RuntimeError("WebSocket session already stopped")
        await self.connection.send(payload)

    def send_message(self, message: PyMessage) -> None:
        if self.bg_loop is None:
            raise RuntimeError("Background event loop is not ready")
        payload = message.to_binary_bytes()

        async def _send() -> None:
            await self._send_bytes(payload)

        future = asyncio.run_coroutine_threadsafe(_send(), self.bg_loop)
        try:
            future.result(timeout=5)
            logger.info("Sent message via WebSocket: %s", message)
        except Exception as exc:  # noqa: BLE001
            logger.error("Failed to send message: %s", exc)
            raise

    def create(self, server: str, port: int, path: str, user_id: int | None = None) -> None:
        logger.info("Creating WebSocket background thread")

        loop_future: asyncio.Future = asyncio.Future()
        start_event = threading.Event()

        def start_loop(future: asyncio.Future) -> None:
            loop = asyncio.new_event_loop()
            asyncio.set_event_loop(loop)
            future.set_result(loop)
            start_event.set()
            loop.run_forever()

        thread = threading.Thread(target=start_loop, args=(loop_future,), daemon=True)
        thread.start()
        if not start_event.wait(timeout=1.0):
            raise TimeoutError("Failed to start background loop")

        loop: asyncio.AbstractEventLoop = loop_future.result()
        coroutine = self.connect(server, port, path, user_id)
        self._connect_future = asyncio.run_coroutine_threadsafe(coroutine, loop)

        self.bg_thread = thread
        self.bg_loop = loop

    def stop_loop_thread(self) -> None:
        logger.info("Stopping WebSocket client thread ...")
        if self.bg_loop is None:
            return

        try:
            asyncio.run_coroutine_threadsafe(self.disconnect(), self.bg_loop).result(timeout=5)
        except Exception as exc:  # noqa: BLE001
            logger.error("Error during disconnect: %s", exc)

        if self._connect_future is not None:
            self._connect_future.cancel()
            self._connect_future = None

        try:
            asyncio.run_coroutine_threadsafe(self._shutdown_loop(), self.bg_loop).result(timeout=5)
        except Exception as exc:  # noqa: BLE001
            logger.error("Error shutting down loop: %s", exc)

        self.bg_loop.call_soon_threadsafe(self.bg_loop.stop)
        if self.bg_thread is not None:
            self.bg_thread.join(timeout=2)

        self.bg_thread = None
        self.bg_loop = None

    async def disconnect(self) -> None:
        self.stop_flag.set()
        if self._receive_task is not None:
            self._receive_task.cancel()
            self._receive_task = None

        await self._cleanup_connection()

    async def _cleanup_connection(self) -> None:
        if self.connection is not None:
            try:
                await self.connection.close(code=1000, reason="client shutdown")
            except Exception as exc:  # noqa: BLE001
                logger.debug("Error while closing WebSocket: %s", exc)

        self.connection = None
        self.connected = False

    async def _shutdown_loop(self) -> None:
        pending = [task for task in asyncio.all_tasks() if task is not asyncio.current_task()]
        for task in pending:
            task.cancel()
        if pending:
            await asyncio.gather(*pending, return_exceptions=True)

    def _build_uri(self, server: str, port: int, path: str) -> str:
        scheme = "wss" if self._secure else "ws"
        normalised_path = path if path.startswith("/") else f"/{path}"
        return f"{scheme}://{server}:{port}{normalised_path}"

    def _build_ssl_context(self) -> ssl.SSLContext:
        context = ssl.create_default_context()
        context.check_hostname = False
        context.verify_mode = ssl.CERT_NONE
        return context
